package com.example.job.factory;

import com.example.cdc.annotation.CdcListener;
import com.example.cdc.listener.CdcMessageListener;
import com.example.job.listener.SysJobListener;
import com.example.job.po.SysJob;
import com.example.job.quartz.Job;
import com.example.job.service.SysJobService;
import com.example.job.utils.CronExpression;
import com.example.latch.MySqlLatchUtils;
import com.example.latch.MysqlLatchListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.ApplicationContext;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author chentao
 * @version 1.0
 * @description: TODO
 * @date 2025/3/26 下午11:07
 */
@CdcListener(tableName = "sys_job")
public class SysJobFactory implements InitializingBean, CommandLineRunner, DisposableBean, CdcMessageListener {

    Logger logger = LoggerFactory.getLogger(getClass());

    private volatile boolean isRunning = true;

    private static final Object object = new Object();

    private volatile long nextTime;

    private static final String sql = "select * from sys_lock where id =3 for update";

    private ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;

    private Map<Long, SysJob> jobMap = new ConcurrentHashMap<>();

    @Resource
    private DataSource dataSource;

    @Resource
    private SysJobService sysJobService;

    @Resource
    private SysJobListener sysJobListener;

    @Resource
    private ApplicationContext applicationContext;

    // 5秒的容错时间
    private static final long INTERVAL = 5000L;

    // 过期时间 默认1分钟
    private static final long COUNT_INTERVAL = 60000L;

    @Override
    public void afterPropertiesSet() throws Exception {
        this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(10);
    }

    @Override
    public void run(String... args) throws Exception {
        MySqlLatchUtils.latchInAction(this.dataSource, sql, new MysqlLatchListener() {
            @Override
            public void isLeader() {
                logger.warn("job worker started ...");
                while (isRunning) {
                    synchronized (object) {
                        long time = nextTime - System.currentTimeMillis();
                        if (time > 0) {
                            try {
                                object.wait(time);
                            } catch (InterruptedException e) {
                                logger.warn("thread wait interrupted : {}", e.getMessage());
                            }
                        }
                        try {
                            Date date = executeBatch();
                            Date nextTime = sysJobService.selectNextRunTime(date);
                            setNextTime(nextTime);
                        } catch (Exception e) {
                            logger.error("job scheduler execute error", e.getCause());
                        }
                    }
                }
            }
        });
    }

    public Date executeBatch() throws Exception {
        Date now = new Date(System.currentTimeMillis() + INTERVAL);
        List<SysJob> sysJobs = sysJobService.selectJobList(now);
        for (SysJob sysJob : sysJobs) {
            if (filterMissJob(sysJob, now)) {
                Date nextTime = sysJob.getNextRunTime();
                if (nextTime == null) {
                    nextTime = now;
                }
                CronExpression cronExpression = new CronExpression(sysJob.getCron());
                Date nextValidTimeAfter = cronExpression.getNextValidTimeAfter(now);
                if (nextValidTimeAfter == null) {
                    sysJob.setStatus(false);
                    sysJob.setNextRunTime(null);
                } else {
                    sysJob.setNextRunTime(nextValidTimeAfter);
                }
                sysJobService.updateById(sysJob);
                this.executeJob(sysJob, nextTime);
            }
        }
        return now;
    }

    public void executeJob(SysJob sysJob, Date runTime) {
        if (sysJob.isCanceled() || jobMap.containsKey(sysJob.getId())) {
            logger.warn("job is skip , jobId: " + sysJob.getId());
        } else {
            jobMap.put(sysJob.getId(), sysJob);
            JobRunningProxy jobRunningProxy = new JobRunningProxy(sysJob, runTime);
            long time = runTime.getTime() - System.currentTimeMillis();
            if (time > 0) {
                this.scheduledThreadPoolExecutor.schedule(jobRunningProxy, time, TimeUnit.MILLISECONDS);
            } else {
                this.scheduledThreadPoolExecutor.execute(jobRunningProxy);
            }
        }
    }

    public void setNextTime(Date nextTime) {
        synchronized (object) {
            this.nextTime = nextTime.getTime();
        }
    }

    public boolean filterMissJob(SysJob sysJob, Date runTime) throws Exception {
        if (sysJob.getNextRunTime() == null) {
            sysJob.setNextRunTime(null);
            sysJob.setStatus(false);
            this.sysJobService.updateById(sysJob);
            return false;
        } else {
            long time = runTime.getTime() - sysJob.getNextRunTime().getTime();
            if (Math.abs(time) < COUNT_INTERVAL) {
                return true;
            } else {
                CronExpression cronExpression = new CronExpression(sysJob.getCron());
                sysJob.setNextRunTime(cronExpression.getNextValidTimeAfter(runTime));
                this.sysJobService.updateById(sysJob);
                return false;
            }
        }
    }

    @Override
    public void destroy() throws Exception {
        this.isRunning = false;
        this.scheduledThreadPoolExecutor.shutdownNow();
    }

    @Override
    public void onCdcMessage(String tableName, String type) {
        synchronized (object) {
            object.notify();
        }
    }

    //任务执行包装类
    class JobRunningProxy implements Runnable {

        private SysJob sysJob;

        private Date runTime;

        public JobRunningProxy(SysJob sysJob, Date runTime) {
            this.sysJob = sysJob;
            this.runTime = runTime;
        }

        @Override
        public void run() {
            Class<? extends Job> jobClass = null;
            try {
                if (sysJob.isCanceled()) {
                    logger.warn("job is canceled and skip , jobId: " + sysJob.getId());
                    return;
                }
                jobClass = (Class<? extends Job>) Class.forName(sysJob.getJobClass());
                Job job = jobClass.newInstance();
                job.exec(applicationContext, sysJob);
                sysJobListener.success(applicationContext, sysJob);
            } catch (Exception e) {
                sysJobListener.failure(applicationContext, sysJob);
                logger.warn("job exec: " + sysJob + " 运行失败");
            } finally {
                if (sysJob.isAutoDelete()){
                    sysJobService.removeById(sysJob.getId());
                }
                jobMap.remove(sysJob.getId());
            }
        }
    }
}
